/**
* MagicInvoker.java 2020/5/16 1:58 下午
* Copyright ©2020 www.bmsoft.com.cn All rights reserved.
* PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
*/
package protocol.magic;

import common.constants.CommonConstants;
import common.exception.MagicRpcException;
import common.exception.RemotingException;
import common.exception.TimeoutException;
import common.model.AsyncRpcResult;
import common.model.FutureContext;
import common.model.Invocation;
import common.model.Invoker;
import common.model.MagicUrl;
import common.model.Result;
import common.model.RpcInvocation;
import common.util.AtomicPositiveInteger;
import common.util.RpcUtils;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.locks.ReentrantLock;
import protocol.AbstractInvoker;
import remoting.exchange.ExchangeClient;

/**
 * File：MagicInvoker.java<br>
 * Title: <br>
 * Description: <br>
 * Company: www.treefinance.com.cn <br>
 * @author heyouchi
 * Date: 2020/5/16
 */
public class MagicInvoker<T> extends AbstractInvoker<T> {

  private final ExchangeClient[] clients;

  private final AtomicPositiveInteger index = new AtomicPositiveInteger();

  private final String version;

  private final ReentrantLock destroyLock = new ReentrantLock();

  private final Set<Invoker<?>> invokers;

  public MagicInvoker(Class<T> serviceType, MagicUrl url, ExchangeClient[] clients) {
    this(serviceType, url, clients, null);
  }

  public MagicInvoker(Class<T> serviceType, MagicUrl url, ExchangeClient[] clients, Set<Invoker<?>> invokers) {
    super(serviceType, url, new String[]{CommonConstants.INTERFACE_KEY, CommonConstants.GROUP_KEY, CommonConstants.TOKEN_KEY, CommonConstants.TIMEOUT_KEY});
    this.clients = clients;
    // get version.
    this.version = url.getParameter(CommonConstants.VERSION_KEY, "0.0.0");
    this.invokers = invokers;
  }

  @Override
  protected Result doInvoke(final Invocation invocation) throws Throwable {
    RpcInvocation inv = (RpcInvocation) invocation;
    final String methodName = RpcUtils.getMethodName(invocation);
    inv.setAttachment(CommonConstants.PATH_KEY, getUrl().getPath());
    inv.setAttachment(CommonConstants.VERSION_KEY, version);

    ExchangeClient currentClient;
    if (clients.length == 1) {
      currentClient = clients[0];
    } else {
      currentClient = clients[index.getAndIncrement() % clients.length];
    }
    try {
      boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
      int timeout = getUrl().getMethodPositiveParameter(methodName, CommonConstants.TIMEOUT_KEY, CommonConstants.DEFAULT_TIMEOUT);
      if (isOneway) {
        boolean isSent = getUrl().getMethodParameter(methodName, CommonConstants.SENT_KEY, false);
        currentClient.sendMessage(inv, isSent);
        return AsyncRpcResult.newDefaultAsyncResult(invocation);
      } else {
        AsyncRpcResult asyncRpcResult = new AsyncRpcResult(inv);
        CompletableFuture<Object> responseFuture = currentClient.request(inv, timeout);
        asyncRpcResult.subscribeTo(responseFuture);
        FutureContext.getContext().setCompatibleFuture(responseFuture);
        return asyncRpcResult;
      }
    } catch (TimeoutException e) {
      throw new MagicRpcException(MagicRpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
    } catch (RemotingException e) {
      throw new MagicRpcException(MagicRpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
    }
  }

  @Override
  public boolean isAvailable() {
    if (!super.isAvailable()) {
      return false;
    }
    for (ExchangeClient client : clients) {
      if (client.isConnected() && !client.hasAttribute(CommonConstants.CHANNEL_ATTRIBUTE_READONLY_KEY)) {
        //cannot write == not Available ?
        return true;
      }
    }
    return false;
  }

  @Override
  public void destroy() {
    // in order to avoid closing a client multiple times, a counter is used in case of connection per jvm, every
    // time when client.close() is called, counter counts down once, and when counter reaches zero, client will be
    // closed.
    if (super.isDestroyed()) {
      return;
    } else {
      // double check to avoid dup close
      destroyLock.lock();
      try {
        if (super.isDestroyed()) {
          return;
        }
        super.destroy();
        if (invokers != null) {
          invokers.remove(this);
        }
        for (ExchangeClient client : clients) {
          try {
            client.close(CommonConstants.DEFAULT_SERVER_SHUTDOWN_TIMEOUT);
          } catch (Throwable t) {
            logger.warn(t.getMessage(), t);
          }
        }

      } finally {
        destroyLock.unlock();
      }
    }
  }
}
